package cn._51doit.live.jobs;

import cn._51doit.live.pojo.DataBean;
import cn._51doit.live.pojo.GiftBean;
import cn._51doit.live.sources.MySQLSource;
import cn._51doit.live.udf.GiftCountFunction;
import cn._51doit.live.udf.JsonToBeanFunction;
import cn._51doit.live.udf.JsonToGiftBeanFunction;
import cn._51doit.live.utils.Constants;
import cn._51doit.live.utils.FlinkUtils;
import cn._51doit.live.utils.FlinkUtilsV2;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.*;

/**
 * 礼物多维度统计
 *
 * 使用广播状态，将维度数据（礼物表中的数据）广播
 */
public class GiftCountV2 {


    public static void main(String[] args) throws Exception {

        ParameterTool parameterTool = ParameterTool.fromPropertiesFile(args[0]);

        DataStream<String> giftLineStream = FlinkUtilsV2.createKafkaStream(parameterTool, parameterTool.getRequired("gift.topic"), SimpleStringSchema.class);
        SingleOutputStreamOperator<GiftBean> giftBeanStream = giftLineStream.process(new JsonToGiftBeanFunction());

        //将维度数据进行广播

        //定义广播状态(OperatorState)
        MapStateDescriptor<Integer, Tuple2<String, Double>> stateDescriptor = new MapStateDescriptor<>("gift-state", TypeInformation.of(Integer.class), TypeInformation.of(new TypeHint<Tuple2<String, Double>>() {}));

        //将维度流进行广播
        BroadcastStream<GiftBean> broadcastStream = giftBeanStream.broadcast(stateDescriptor);

        //读取事实流（用户行为数据）
        //读取行为数据
        DataStream<String> lines = FlinkUtilsV2.createKafkaStream(parameterTool, parameterTool.getRequired("even.topic"), SimpleStringSchema.class);

        SingleOutputStreamOperator<DataBean> beanStream = lines.process(new JsonToBeanFunction());

        SingleOutputStreamOperator<DataBean> filtered = beanStream.filter(bean -> Constants.LIVE_REWARD.equals(bean.getEventId()));

        //先将用户的行为数据按照主播ID进行key
        KeyedStream<DataBean, String> keyedStream = filtered.keyBy(bean -> bean.getProperties().get("anchor_id").toString());

        //将行为日志流（事实）和礼物流（维度流）进行connect
        //先keyBy再与广播流进行connect，在关联广播维度流的同时，可以使用keyedState进行聚合
        SingleOutputStreamOperator<Tuple3<String, Integer, Double>> res = keyedStream.connect(broadcastStream).process(new GiftCountFunction(stateDescriptor));

        //将结果写入到Redis中
        res.print();

        FlinkUtilsV2.env.execute();

    }
}
